package team.bluepen.supermarket.data.dao;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;
import team.bluepen.supermarket.constant.Constant;
import team.bluepen.supermarket.data.dao.abstraction.MarketDatabase;
import team.bluepen.supermarket.data.entity.Good;
import team.bluepen.supermarket.data.entity.MarketTable;
import team.bluepen.supermarket.data.hbase.GoodsLineReader;
import team.bluepen.supermarket.util.StringUtil;

import java.io.IOException;

/**
 * 操作{@link Good} 实体类对应表
 * @author Kai
 */
public class GoodDatabase extends MarketDatabase<Good> {
    private static final String TAG = "GoodDatabase";
    private static final MarketTable PRODUCT_TABLE = new MarketTable()
            .setName(Constant.GoodTable.TABLE_NAME)
            .setFamilies(new String[]{Constant.GoodTable.TABLE_FAMILY_2,
                    Constant.GoodTable.TABLE_FAMILY_1});

    public GoodDatabase() throws IOException {
        super(PRODUCT_TABLE);
    }

    @Override
    protected Put buildPut(Good good) {
        GoodsLineReader reader = new GoodsLineReader();
        return reader.buildPut(good);
    }

    @Override
    protected Good getInformation(Result result) {
        Good good = new Good();
        for (Cell cell : result.rawCells()) {
            String row = Bytes.toString(CellUtil.cloneRow(cell));
            String qualifier = Bytes.toString(CellUtil.cloneQualifier(cell));
            byte[] value = CellUtil.cloneValue(cell);
            convertHelper(good, row, qualifier, value);
        }
        return good;
    }

    private void convertHelper(Good good, String row, String qualifier, byte[] value) {
        if (good == null || StringUtil.isEmpty(row)){
            return;
        }
        good.setRow(row);
        if (Constant.GoodTable.QUALIFIER_ID.equals(qualifier)) {
            good.setId(Bytes.toLong(value));
        }
        if (Constant.GoodTable.QUALIFIER_NAME.equals(qualifier)) {
            good.setName(Bytes.toString(value));
        }
        if (Constant.GoodTable.QUALIFIER_STATE.equals(qualifier)) {
            good.setState(Bytes.toString(value));
        }
        if (Constant.GoodTable.QUALIFIER_AMOUNT.equals(qualifier)) {
            good.setAmount(Bytes.toInt(value));
        }
        if (Constant.GoodTable.QUALIFIER_PRICE.equals(qualifier)) {
            good.setPrice(Bytes.toInt(value));
        }
    }

}
